/*
Copyright 2010 Dmitry Barashev

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
 */
package net.barashev.seclub.io;

import index.CalculatorOfWorkingWithDisk;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ReadableByteChannel;

/**
 * A slot for sequential iterating over a continuous array of bytes using some fixed
 * number of buffers.
 * Before the first call of pull() the slot is in undetermined state.
 * Iteration is one-way and cannot be rewinded. If iteration reaches the end of
 * the input data, the slot releases itself automatically. If iteration is interrupted
 * before it reaches the end of data, the iterating code must release this slot.
 * @author Dmitry Barashev
 */
public class IteratingSlot<T, M extends DataMarshaller<T>> implements Slot {
    private ByteBuffer slice;
    private final ReadableByteChannel channel;
    private final BufferManager bufferManager;
    private final M marshaller;
    private int read;
    private boolean isReleased;

    IteratingSlot(ByteBuffer slice, ReadableByteChannel channel, M marshaller, BufferManager mgr) {
        this.slice = slice;
        this.channel = channel;
        this.bufferManager = mgr;
        this.marshaller = marshaller;
        this.isReleased = false;
    }

    /**
     * Pulls the next chunk of data into the slot.
     * @return {@true} if there was data to pull.
     * @throws IOException
     */
    public boolean pull() throws IOException {
        this.slice.clear();
        this.read = this.channel.read(this.slice);
        if (this.read == -1) {
            release();
        } else {
            CalculatorOfWorkingWithDisk.callPull(getBlockCount());
        }
        return this.read > 0;
    }

    /**
     * @return the number of blocks fetched by the last call of pull()
     */
    public int getBlockCount() {
        int dataLength =Math.min(this.read, this.slice.capacity());
        return dataLength / this.bufferManager.getBlockSize() + (dataLength % this.bufferManager.getBlockSize() > 0 ? 1 : 0);
    }

    /**
     * Returns one of the blocks currently pulled into the slot.
     * Precondition: idx>=0 and idx<getBlockCount()
     * @param idx Index of the requested block.
     * @return a new instance of block object.
     */
    public FixedLengthRecordBlock<T, DataMarshaller<T>> getBlock(final int idx) {

        int relativeOffset = idx * this.bufferManager.getBlockSize();
        int blockSize = Math.min(this.bufferManager.getBlockSize(), this.read - relativeOffset);
        return new FixedLengthRecordBlock<T, DataMarshaller<T>>(
                this.slice.array(),
                blockSize, this.marshaller) {
            @Override
            protected int getBlockStartOffset() {
                return IteratingSlot.this.slice.arrayOffset() + idx * IteratingSlot.this.bufferManager.getBlockSize();
            }
        };
    }

    /**
     * @return the total number of data records in this slot.
     */
    public int getRecordCount() {
        return this.read / this.marshaller.getDataItemSize();
    }

    public ByteBuffer getSlice() {
        return this.slice;
    }

    /**
     * Releases this slot and its buffers
     */
    public void release() {
    	CalculatorOfWorkingWithDisk.callReleaseIteratingSlot();
    	
        this.isReleased = true;
        try {
            this.channel.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public boolean isReleased() {
        return this.isReleased;
    }

    public void setSlice(ByteBuffer slice) {
        this.slice = slice;
    }
}